package com.dc.rxjava;

import android.util.Log;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Predicate;

/**
 * 条件操作符
 *
 * @author 冰柠檬dc
 */
public class ConditionOperatorUtils {

    private static final String TAG = "RxJava";

    /**
     * all操作符：判断事件序列是否全部满足某个事件，如果都满足则返回 true，反之则返回 false。
     * any操作符：判断事件序列是否有一个足某个事件，如果有一个满足则返回 true，反之则返回 false。
     *
     * @return Disposable
     */
    public static Disposable all() {
        return Observable.just(1, 2, 3).all(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer < 5;
            }
        }).subscribe(aBoolean -> {
            Log.e(TAG, "onNext:" + aBoolean);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:true
    }

    /**
     * takeWhile操作符：发射Observable的数据，直到一个指定的条件不成立。(一直take，直到某个条件不满足时直接发送onComplete)
     * TODO 注意和过滤操作符filter、条件操作符takeUntil、skipWhile、skipUntil区别
     *
     * @return Disposable
     */
    public static Disposable takeWhile() {
        return Observable.just(1, 4, 3, 5, 2, 6).takeWhile(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer <= 4;
            }
        }).subscribe(integer -> {
            Log.e(TAG, "onNext:" + integer);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(Observable发射数据，不满足条件后直接onComplete)
        // onSubscribe
        // onNext:1
        // onNext:4
        // onNext:3
        // onComplete
    }

    /**
     * takeUntil操作符（2种用法）：
     * 1.当事件满足此条件时，下一次的事件就不会被发送了。
     * 2.发射原始Observable数据，直到第二个Observable发射了一项数据后，原始Observable后面的全部丢弃
     * TODO 注意和过滤操作符filter、条件操作符takeWhile、skipWhile、skipUntil区别
     *
     * @return Disposable
     */
    public static Disposable takeUntil() {
//        return Observable.just(1, 4, 3, 5, 2, 6).takeUntil(new Predicate<Integer>() {
//            @Override
//            public boolean test(Integer integer) throws Exception {
//                return integer <= 4;
//            }
//        }).subscribe(integer -> {
//            Log.e(TAG, "onNext:" + integer);
//        }, throwable -> {
//            Log.e(TAG, "onError");
//        }, () -> {
//            Log.e(TAG, "onComplete");
//        }, onSubscribe -> {
//            Log.e(TAG, "onSubscribe");
//        });
        // 打印结果：(Observable发射数据，满足条件后下一次直接onComplete)
        // onSubscribe
        // onNext:1
        // onComplete

        return Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS).takeUntil(Observable.intervalRange(6, 5, 3, 1, TimeUnit.SECONDS)).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(第一个Observable发射数据，第二个Observable发射数据后第一个Observable直接onComplete)
        // onSubscribe
        // onNext:1
        // onNext:2
        // onNext:3
        // onComplete
    }

    /**
     * skipWhile操作符：丢弃原始Observable发射的数据，直到一个指定的条件不成立。(一直skip，直到某个条件不满足时take后面所有的事件)
     * TODO 注意和过滤操作符filter、条件操作符takeWhile、takeUntil、skipUntil区别
     *
     * @return Disposable
     */
    public static Disposable skipWhile() {
        return Observable.just(1, 4, 3, 5, 2, 6).skipWhile(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer <= 4;
            }
        }).subscribe(integer -> {
            Log.e(TAG, "onNext:" + integer);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(满足时一直skip，丢弃原始Observable发射的数据，不满足条件后获取剩下所有的事件)
        // onSubscribe
        // onNext:5
        // onNext:2
        // onNext:6
        // onComplete
    }

    /**
     * skipUntil操作符：丢弃原始Observable发射的数据，直到第二个Observable发射了一项数据。
     * TODO 注意和过滤操作符filter、条件操作符takeWhile、takeUntil、skipWhile区别
     *
     * @return Disposable
     */
    public static Disposable skipUntil() {
        return Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS).skipUntil(Observable.intervalRange(6, 5, 3, 1, TimeUnit.SECONDS)).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：(丢弃第一个Observable发射数据，直到第二个Observable发射数据后获取第一个Observable剩下所有的事件)
        // onSubscribe
        // onNext:4
        // onNext:5
        // onComplete
    }

    /**
     * sequenceEqual操作符：判断两个 Observable 发送的事件是否相同。
     *
     * @return Disposable
     */
    public static Disposable sequenceEqual() {
        return Observable.sequenceEqual(Observable.just(1, 2, 3), Observable.just(1, 2, 3)).subscribe(aBoolean -> {
            Log.e(TAG, "onNext:" + aBoolean);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:true
    }

    /**
     * contains操作符：判断事件序列中是否含有某个元素，如果有则返回 true，如果没有则返回 false。
     *
     * @return Disposable
     */
    public static Disposable contains() {
        return Observable.just(1, 2, 3).contains(3).subscribe(aBoolean -> {
            Log.e(TAG, "onNext:" + aBoolean);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:true
    }

    /**
     * isEmpty操作符：判断事件序列是否为空。
     *
     * @return Disposable
     */
    public static Disposable isEmpty() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onComplete();
            }
        }).isEmpty().subscribe(aBoolean -> {
            Log.e(TAG, "onNext:" + aBoolean);
        }, throwable -> {
            Log.e(TAG, "onError");
        });
        // 打印结果：
        // onNext:true
    }

    /**
     * amb操作符：amb() 要传入一个 Observable 集合，但是只会发送最先发送事件的 Observable 中的事件，其余 Observable 将会被丢弃。
     *
     * @return Disposable
     */
    public static Disposable amb() {
        ArrayList<Observable<Long>> list = new ArrayList<>();

        list.add(Observable.intervalRange(1, 3, 2, 1, TimeUnit.SECONDS));
        list.add(Observable.intervalRange(6, 3, 0, 1, TimeUnit.SECONDS));
        return Observable.amb(list).subscribe(aBoolean -> {
            Log.e(TAG, "onNext:" + aBoolean);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:6
        // onNext:7
        // onNext:8
        // onComplete
    }

    /**
     * defaultIfEmpty操作符：如果观察者只发送一个 onComplete() 事件，则可以利用这个方法发送一个值
     *
     * @return Disposable
     */
    public static Disposable defaultIfEmpty() {
        return Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onComplete();
            }
        }).defaultIfEmpty(6).subscribe(i -> {
            Log.e(TAG, "onNext:" + i);
        }, throwable -> {
            Log.e(TAG, "onError");
        }, () -> {
            Log.e(TAG, "onComplete");
        }, onSubscribe -> {
            Log.e(TAG, "onSubscribe");
        });
        // 打印结果：
        // onSubscribe
        // onNext:6
        // onComplete
    }
}
